package com.talkyun.utils.rate;

import com.google.common.util.concurrent.RateLimiter;
import com.talkyun.utils.ExecutorBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class RateBaseExecutor {
    private static final RateBaseExecutor INSTANCE = new RateBaseExecutor();
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    private Map<String, RateLimiter> mapping = new ConcurrentHashMap<String, RateLimiter>();
    private ExecutorService es;

    public static void addRateRule(String limitKey, double tps) {
        addRateRule(limitKey, tps, 1.2);
    }

    public static void addRateRule(String limitKey, double tps, double warmUpSeconds) {
        RateLimiter rate = RateLimiter.create(tps, (long) warmUpSeconds * 1000, MILLISECONDS);
        INSTANCE.mapping.put(limitKey, rate);
    }

    public static RateBaseExecutor get() {
        return INSTANCE;
    }

    private RateBaseExecutor() {
        int coreSize = 4;
        int maxiSize = 16;
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(coreSize / 2);
        es = new ExecutorBuilder().prefix("openx-rl").minPoolSize(coreSize).maxPoolSize(maxiSize).queue(queue).build();
    }

    // limit
    // key (service.method) : Limiter
    public void execute(final String limitKey, final Runnable task) {
        es.execute(new Runnable() {
            @Override
            public void run() {
                RateLimiter limiter = mapping.get(limitKey);
                if (limiter == null) {
                    task.run();
                } else {
                    double time = limiter.acquire();
                    task.run();
                }
            }
        });
    }

    public static void main(String[] args) throws InterruptedException {
        RateBaseExecutor.addRateRule("/oto/mt/listChannelSku", 2);
        RateBaseExecutor rbe = RateBaseExecutor.get();
        final AtomicLong no = new AtomicLong(10);
        Thread.sleep(1000);

        for (int i = 0; i < 15; i++) {
            rbe.execute("/oto/mt/listChannelSku", new Runnable() {
                @Override
                public void run() {
                    long num = no.incrementAndGet();
                    System.out.println(num + "  " + Thread.currentThread().getName() + " : " + System.currentTimeMillis() + " -> hello");
                    throw new RuntimeException("hh");
                }
            });

            // System.out.println(":" + i);
        }

        TimeUnit.MINUTES.sleep(5);
    }
}